package dacp.etl.kafka.hdfs.connect.partitioner;

import java.util.Locale;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang.StringUtils;
import org.joda.time.DateTimeZone;

import io.confluent.connect.hdfs.HdfsSinkConnectorConfig;
import io.confluent.connect.hdfs.partitioner.TimeBasedPartitioner;

public class HourlyPartitioner extends TimeBasedPartitioner {

	private static long partitionDurationMs = TimeUnit.HOURS.toMillis(1);
	// private static String pathFormat =
	// "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH/";
	private static String pathFormat = "'month_id'=YYYYMM/'day_id'=YYYYMMdd/'hour_id'=YYYYMMddHH/";

	private static String localeString = "zh";
	private static String timeZoneString = "Asia/Shanghai";

	@Override
	public void configure(Map<String, Object> config) {
		String pathFormatS = (String) config.get(HdfsSinkConnectorConfig.PATH_FORMAT_CONFIG);
		if (StringUtils.isNotEmpty(pathFormatS)) {
			pathFormat = pathFormatS;
		}

		String hiveIntString = (String) config.get(HdfsSinkConnectorConfig.HIVE_INTEGRATION_CONFIG);
		boolean hiveIntegration = hiveIntString != null && hiveIntString.toLowerCase().equals("true");
		Locale locale = new Locale(localeString);
		DateTimeZone timeZone = DateTimeZone.forID(timeZoneString);
		init(partitionDurationMs, pathFormat, locale, timeZone, hiveIntegration);
	}

	@Override
	public String generatePartitionedPath(String topic, String encodedPartition) {
		return encodedPartition;
	}

	public String getPathFormat() {
		return pathFormat;
	}
}